//
// Created by ole on 24-8-18.
//

#include "IPCStrategyLocal.h"

#include <unistd.h>

#include "../../utils/Log.h"

bool IPCStrategyLocal::start() {
    std::lock_guard<std::mutex> lock(memberMutex);
    if (isStated == true) {
        Log::e(TAG, "Already started.");
        return false;
    }
    isStated = true;
    eventThread = std::thread(&IPCStrategyLocal::eventLoop, this);
    Log::d(TAG, "IPCStrategyLocal has started.");
    return true;
}

bool IPCStrategyLocal::stop() {
    std::lock_guard<std::mutex> lock(memberMutex);
    if (isStated != true) {
        Log::e(TAG, "Already stopped.");
        return false;
    }
    isStated = false;
    eventThread.join();
    Log::d(TAG, "IPCStrategyLocal has stopped.");
    return true;
}

void IPCStrategyLocal::subscribe(const std::shared_ptr<Topic> &topic,
                                 const std::shared_ptr<MessageReceiver> &receiver) {
    if (receiver == nullptr || topic == nullptr) {
        Log::e(TAG, "Invalid arguments.");
        return;
    }

    std::lock_guard<std::mutex> lock(receiverMutex);
    receiverSetMap[*topic].insert(receiver);
    // dumpTopicReceivers();
    Log::d(TAG, "Subscribe receiver for '" + topic->getAction() + "'.");
}

void IPCStrategyLocal::unsubscribe(const std::shared_ptr<Topic> &topic,
                                   const std::shared_ptr<MessageReceiver> &receiver) {
    if (receiver == nullptr || topic == nullptr) {
        Log::e(TAG, "Invalid arguments.");
        return;
    }

    std::lock_guard<std::mutex> lock(receiverMutex);
    receiverSetMap[*topic].erase(receiver);
    Log::d(TAG, "UnSubscribe receiver for '" + topic->getAction() + "'.");
}

void IPCStrategyLocal::publishMessage(const std::shared_ptr<Message> &msg) {
    Log::d(TAG, "Publish message '" + msg->getAction() + "'.");
    std::lock_guard<std::mutex> lock(eventMutex);
    eventQueue.push(msg);
}

void IPCStrategyLocal::loopForever() {
    Log::d(TAG, "IPCStrategyLocal loop forever.");
    eventThread.join();
    Log::d(TAG, "IPCStrategyLocal loop exited.");
}

void IPCStrategyLocal::eventLoop() {
    Log::d(TAG, "eventLoop() enter.");
    while (isStated) {
        usleep(10 * 1000);
        std::lock_guard<std::mutex> lock(eventMutex);
        if (!eventQueue.empty()) {
            const std::shared_ptr<Message> intent = eventQueue.top();
            eventQueue.pop();
            handleEvent(intent);
        }
    }

    Log::d(TAG, "eventLoop() exit.");
}

void IPCStrategyLocal::handleEvent(const std::shared_ptr<Message> &msg) {
    // Log::d(TAG, "HandleEvent(): '" + msg->toString() + "'.");
    const std::chrono::microseconds warningDuration(2 * 1000);
    const auto start = std::chrono::high_resolution_clock::now();

    std::lock_guard<std::mutex> lock(receiverMutex);
    // dumpTopicReceivers();
    const ReceiverSet receivers = receiverSetMap[Topic(msg->getAction())];
    for (const auto &receiver: receivers) {
        if (receiver) {
            receiver->onReceived(msg);
        }
    }

    const auto end = std::chrono::high_resolution_clock::now();
    const auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);
    if (duration > warningDuration) {
        std::stringstream ss;
        ss << "Took " << duration.count() << " microseconds on event " << msg->getAction();
        Log::w(TAG, ss.str());
    }
}

void IPCStrategyLocal::dumpTopicReceivers() const {
    for (const auto &pair: receiverSetMap) {
        std::ostringstream ss;
        ss << "Dump Topic: " << pair.first.getAction() << ", Receivers: {";
        for (const auto &receiver: pair.second) {
            ss << (&receiver) << " ";
        }
        ss << "}";
        Log::d(TAG, ss.str());
    }
}
